package com.zhisheng.sql.blink.stream.example;

import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.util.*;
import java.util.stream.Collectors;

public class UserSimilarityFunction implements 
    WindowFunction<Row, Tuple4<Integer, Integer, Double, List<String>>, Integer, TimeWindow> {

    @Override
    public void apply(
            Integer key,
            TimeWindow window,
            Iterable<Row> users,
            Collector<Tuple4<Integer, Integer, Double, List<String>>> out) {

        List<UserProfile> userProfiles = new ArrayList<>();
        
        // 1. 转换用户数据为UserProfile对象
        for (Row row : users) {
            Integer userId = (Integer) row.getField(0);
            String username = (String) row.getField(1);
            Double spendingLevel = (Double) row.getField(2);
            String[] categories = ((String) row.getField(3)).split(",");
            
            userProfiles.add(new UserProfile(userId, username, spendingLevel, new HashSet<>(Arrays.asList(categories))));
        }

        // 2. 计算用户间的相似度并生成推荐
        for (UserProfile user1 : userProfiles) {
            List<UserSimilarity> similarities = new ArrayList<>();
            
            // 计算当前用户与其他用户的相似度
            for (UserProfile user2 : userProfiles) {
                if (!user1.userId.equals(user2.userId)) {
                    double similarity = calculateSimilarity(user1, user2);
                    similarities.add(new UserSimilarity(user2.userId, similarity));
                }
            }

            // 排序并获取最相似的用户
            similarities.sort((a, b) -> Double.compare(b.similarity, a.similarity));
            
            // 为每个用户生成推荐
            if (!similarities.isEmpty()) {
                UserSimilarity mostSimilar = similarities.get(0);
                List<String> recommendedProducts = getRecommendedProducts(user1.userId, mostSimilar.userId);
                
                out.collect(new Tuple4<>(
                    user1.userId,
                    mostSimilar.userId,
                    mostSimilar.similarity,
                    recommendedProducts
                ));
            }
        }
    }

    private double calculateSimilarity(UserProfile user1, UserProfile user2) {
        // 计算消费等级相似度 (归一化)
        double spendingDiff = Math.abs(user1.spendingLevel - user2.spendingLevel);
        double maxSpending = Math.max(user1.spendingLevel, user2.spendingLevel);
        double spendingSimilarity = 1 - (spendingDiff / maxSpending);

        // 计算类别偏好相似度 (Jaccard相似度)
        Set<String> intersection = new HashSet<>(user1.preferredCategories);
        intersection.retainAll(user2.preferredCategories);
        
        Set<String> union = new HashSet<>(user1.preferredCategories);
        union.addAll(user2.preferredCategories);
        
        double categorySimilarity = union.isEmpty() ? 0 : 
            (double) intersection.size() / union.size();

        // 综合相似度 (这里给予类别偏好更高的权重)
        return (0.3 * spendingSimilarity + 0.7 * categorySimilarity);
    }

    private List<String> getRecommendedProducts(Integer userId, Integer similarUserId) {
        // 在实际应用中，这里应该查询相似用户最近购买的商品
        // 这里简单返回一个示例列表
        try {

            Set<Integer> userProducts = UserPurchaseSetProcessor.userPurchaseMap.get(userId);
            Set<Integer> similarUserProducts = UserPurchaseSetProcessor.userPurchaseMap.get(similarUserId);
            if (userProducts == null){
                userProducts = new HashSet<>();
            }
            if (similarUserProducts == null){
                similarUserProducts = new HashSet<>();
            }
            //求出不在 user 中的id
            Set<Integer> finalUserProducts = userProducts;
            Set<Integer> result = similarUserProducts.stream()
                    .filter(e -> !finalUserProducts.contains(e)) // 过滤掉a中存在的元素
                    .collect(Collectors.toSet());
            return result.stream().map(String::valueOf).collect(Collectors.toList());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    // 内部类：用户画像
    private static class UserProfile {
        Integer userId;
        String username;
        Double spendingLevel;
        Set<String> preferredCategories;

        UserProfile(Integer userId, String username, Double spendingLevel, Set<String> preferredCategories) {
            this.userId = userId;
            this.username = username;
            this.spendingLevel = spendingLevel;
            this.preferredCategories = preferredCategories;
        }
    }

    // 内部类：用户相似度
    private static class UserSimilarity {
        Integer userId;
        Double similarity;

        UserSimilarity(Integer userId, Double similarity) {
            this.userId = userId;
            this.similarity = similarity;
        }
    }
}
